#include "jthreadpool.h"
#include <JXQCore/JThread>
#include <JXQCore/JMutex>
#include <JXQCore/JDebug>

#include <condition_variable>
#include <atomic>
#include <list>
#include <set>

namespace JXQ
{

using PackagedTask = std::function<void()>;

// 任务队列
class J_DECL_HIDDEN TaskQueue
{
public:
	explicit TaskQueue(int m_maxSize, int expiryTimeout);

public:
    void push(std::function<void()> &&task); //生产
	PackagedTask dequeue(); //消费

public:
	inline bool empty() const;
	inline size_t size() const;

public:
	void setMaxQueueLength(int max);
	inline int maxQueueLength() const;
	inline void setExpiryTimeout(uint32_t msecs);
	inline uint32_t expiryTimeout() const;

public:
	void exit(); //队列进入退出模式，唤醒所以在队列上阻塞的线程
	inline void notify(uint32_t num);

private:
	bool wait(std::condition_variable &condition, int msecs = -1);

private:
	mutable std::mutex m_mutex;
	std::list<PackagedTask> m_list;

	std::mutex m_waitMutex;
	std::condition_variable m_producerCondition; //生产者
	std::condition_variable m_consumerCondition; //消费者
	std::atomic_int m_expiryTimeout; //自动销毁时间

	std::atomic_bool m_quitFlag {false}; //析构标志
	std::atomic_int m_notifyCount {0};

	mutable std::mutex m_maxSizeMutex;
	int m_maxSize; //队列长度限制
};

TaskQueue::TaskQueue(int maxSize, int expiryTimeout) :
	m_expiryTimeout(expiryTimeout)
{
	J_MUTEX_LOCKER(m_maxSizeMutex);
	this->m_maxSize = maxSize;
}

void TaskQueue::push(std::function<void()> &&task)
{
	if( m_quitFlag )
		return ;

	size_t maxSize = this->m_maxSize;
	J_MUTEX_LOCKER(m_mutex);
	if( maxSize > 1 and m_list.size() >= maxSize )
	{
		J_MUTEX_LOCKER_UNLOCK();
		if( wait(m_producerCondition) == false )
			return ;
		J_MUTEX_LOCKER_RELOCK();
	}
	m_list.push_back(std::forward<PackagedTask>(task));
	m_consumerCondition.notify_one(); //唤醒一个消费者
}

PackagedTask TaskQueue::dequeue()
{
	if( m_quitFlag )
		return PackagedTask(nullptr);

	J_MUTEX_LOCKER(m_mutex);
	while( m_list.empty() )
	{
		J_MUTEX_LOCKER_UNLOCK();
		if( wait(m_consumerCondition, m_expiryTimeout) == false )
			return PackagedTask(nullptr);
		J_MUTEX_LOCKER_RELOCK();
	}

	auto task = m_list.front();
	m_list.pop_front();
	m_producerCondition.notify_one(); //唤醒一个生产者
	return task;
}

inline bool TaskQueue::empty() const
{
	J_MUTEX_LOCKER(m_mutex);
	return m_list.empty();
}

inline size_t TaskQueue::size() const
{
	J_MUTEX_LOCKER(m_mutex);
	return m_list.size();
}

void TaskQueue::setMaxQueueLength(int max)
{
	J_MUTEX_LOCKER(m_maxSizeMutex);
	if( max > m_maxSize )
	{
		uint32_t count = max - m_maxSize;
		m_maxSize = max;
		J_MUTEX_LOCKER_UNLOCK();
		for(uint32_t i=0; i<count; i++)
			m_producerCondition.notify_one();
	}
	else
		m_maxSize = max;
}

inline int TaskQueue::maxQueueLength() const
{
	J_MUTEX_LOCKER(m_maxSizeMutex);
	return m_maxSize;
}

inline void TaskQueue::setExpiryTimeout(uint32_t msecs)
{
	m_expiryTimeout = msecs;
	m_consumerCondition.notify_all();
}

inline uint32_t TaskQueue::expiryTimeout() const
{
	return m_expiryTimeout;
}

void TaskQueue::exit() //队列进入退出模式，唤醒所以在队列上阻塞的线程
{
	m_list.clear();
	m_quitFlag = true;
	m_producerCondition.notify_all();
	m_consumerCondition.notify_all();
	std::this_thread::yield(); //放弃CPU
}

inline void TaskQueue::notify(uint32_t num)
{
	m_notifyCount += num;
	for(uint32_t i=0; i<num; i++)
		m_consumerCondition.notify_one();
}

bool TaskQueue::wait(std::condition_variable &condition, int msecs)
{
	if( m_quitFlag )
		return false;

	std::unique_lock<std::mutex> locker(m_waitMutex);

	if( msecs > 0 )
	{
		if( condition.wait_for(locker, std::chrono::milliseconds(msecs))
			== std::cv_status::timeout
#ifdef _MSC_VER
			or m_list.empty()
#endif
			) return false;

		else if( m_notifyCount > 0 )
		{
			--m_notifyCount;
			return false;
		}
	}
	else
		condition.wait(locker);

	return m_quitFlag? false : true;
}

/*--------------------------------------------------------------------------------------------------*/

// 线程
class J_DECL_HIDDEN TPThread : public JThread
{
public:
	TPThread(JThreadPoolPirvate *jtpp) :
		m_jtpp(jtpp) { start(); }

private:
	void run() override;
	JThreadPoolPirvate *m_jtpp;
};

/*--------------------------------------------------------------------------------------------------*/

class J_DECL_HIDDEN JThreadPoolPirvate
{
public:
	JThreadPoolPirvate(uint32_t maxCount, int maxQueue, int expiryTimeout);

public:
    void newTask(std::function<void()> &&task);
	void setMaxThreadCount(uint32_t max);

	inline int threadAvailableCount() const
	{ return maxCount - activeCounter; }

	inline int activeThreadCount() const
	{ return activeCounter; }

	inline int threadCount() const
	{ return ref - 1; }

public:
	inline int maxThreadCount() const
	{ return maxCount; }

	inline void setMaxQueueLength(int max) { taskQueue.setMaxQueueLength(max); }
	inline int maxQueueLength() const { return taskQueue.maxQueueLength(); }

	inline void setExpiryTimeout(uint32_t msecs) { taskQueue.setExpiryTimeout(msecs); }
	inline uint32_t expiryTimeout() const { return taskQueue.expiryTimeout(); }

public:
	bool wait(int msecs);

	// 这个类对象不能直接delete，
	// 因为可能有线程没完全结束，
	// 还占用这个类对象，通过计
	// 数器方式延迟析构
	inline void destory();

private:
	friend class TPThread;

	std::atomic_uint activeCounter {0}; //活动线程数量
	std::atomic_uint maxCount; //最大线程数量

	// 原子变量保护的是变量本身，只能保证自操作（如：自增操作）是原子的
	// 多个变量相互操作无法保证原子性:
	// 例如两个变量相减，取/赋值操作是原子的（各自的），减法运算不是原子的（寄存器操作）
	mutable std::mutex MTMutex;

	TaskQueue taskQueue; //任务队列
	std::mutex waitConditionMutex;
	std::condition_variable waitCondition; //等待线程全部结束

	std::atomic_uint ref {1}; //计数器
};

JThreadPoolPirvate::JThreadPoolPirvate
(uint32_t maxCount, int maxQueue, int expiryTimeout) :
	taskQueue(maxQueue, expiryTimeout)
{
	if( maxCount < 1 )
		maxCount = 1;
	this->maxCount = maxCount;
}

void JThreadPoolPirvate::newTask(std::function<void()> &&task)
{
	size_t threadCount = ref - 1;
	size_t freeCount = J_SCT(size_t, threadCount - activeCounter);

	MTMutex.lock();
	if( freeCount <= taskQueue.size() and threadCount < maxCount )
	{
		MTMutex.unlock();
		++ref; // 计数器(原子)加1
		(new TPThread(this))->setEnableDeleteOnFinished();
	}
	else
		MTMutex.unlock();

	taskQueue.push(std::forward<PackagedTask>(task));
}

void JThreadPoolPirvate::setMaxThreadCount(uint32_t max)
{
	if( max < 1 )
		max = 1;

	J_MUTEX_LOCKER(MTMutex);
	if( max < maxCount )
	{
		maxCount = max;
		int num = threadCount() - max;
		if( num > 0 )
			taskQueue.notify(num);
	}
	else
		maxCount = max;
}

bool JThreadPoolPirvate::wait(int msecs)
{
	if( activeCounter <= 0 and taskQueue.empty() )
		return true;
	std::unique_lock<std::mutex> locker(waitConditionMutex);

	if( msecs < 1 )
	{
		waitCondition.wait(locker);
		return true;
	}
	return waitCondition.wait_for(locker, std::chrono::milliseconds(msecs))
			!= std::cv_status::timeout;
}

// 这个类对象不能直接delete，因为可能有线程没完全
// 结束，还占用这个类对象，通过计数器方式延迟析构
inline void JThreadPoolPirvate::destory()
{
	// 队列进入退出模式，唤醒所以在队列上阻塞的线程
	taskQueue.exit();

	wait(-1); //等待任务函数结束
	if( not --ref ) //计数器(原子)为0时，自杀
		delete this;
}

/*--------------------------------------------------------------------------------------------------*/

// 线程函数
void TPThread::run()
{
	for(;;)
	{
		// 取出任务
		auto task = m_jtpp->taskQueue.dequeue();
		if( task == nullptr )
		{
			if( m_jtpp->activeCounter == 0 )
				m_jtpp->waitCondition.notify_all();
			break;
		}

		++m_jtpp->activeCounter; //活动计数器
		task(); //执行任务

		// 顺序不能变
		if( --m_jtpp->activeCounter == 0 and m_jtpp->taskQueue.empty() )
			m_jtpp->waitCondition.notify_all();

		J_MUTEX_LOCKER(m_jtpp->MTMutex);
		if( J_SCT(unsigned int, m_jtpp->threadCount()) > m_jtpp->maxCount )
			break;
	}

	if( not --m_jtpp->ref ) //计数器(原子)为0时，干掉d_ptr
		delete m_jtpp;
}

/*--------------------------------------------------------------------------------------------------*/

// 接口层
JThreadPool::JThreadPool(uint32_t maxCount, int maxQueue, int expiryTimeout) :
	d_ptr(new JThreadPoolPirvate(maxCount, maxQueue, expiryTimeout))
{

}

JThreadPool::~JThreadPool()
{
	// 不能直接delete，因为可能有线程没完全结束，
	// 还占用d_ptr，通过计数器方式延迟析构
	d_ptr->destory();
}

std::future<void> JThreadPool::start(JRunnable *runnable)
{
	return start([runnable]()
	{
		++runnable->counter;
		runnable->run();
		if( --runnable->counter == 0 )
			delete runnable;
	});
}

int JThreadPool::threadAvailableCount() const
{
	return d_ptr->threadAvailableCount();
}

int JThreadPool::activeThreadCount() const
{
	return d_ptr->activeThreadCount();
}

int JThreadPool::threadCount() const
{
	return d_ptr->threadCount();
}

void JThreadPool::setMaxThreadCount(uint32_t max)
{
	d_ptr->setMaxThreadCount(max);
}

int JThreadPool::maxThreadCount() const
{
	return d_ptr->maxThreadCount();
}

void JThreadPool::setMaxQueueLength(int max)
{
	d_ptr->setMaxQueueLength(max);
}

int JThreadPool::maxQueueLength() const
{
	return d_ptr->maxQueueLength();
}

void JThreadPool::setExpiryTimeout(uint32_t msecs)
{
	d_ptr->setExpiryTimeout(msecs);
}

uint32_t JThreadPool::expiryTimeout() const
{
	return d_ptr->expiryTimeout();
}

JThreadPool &JThreadPool::globalInstance()
{
	static JThreadPool pool(100);
	return pool;
}

bool JThreadPool::wait(int msecs)
{
	return d_ptr->wait(msecs);
}

void JThreadPool::yield(std::function<void()> &&task)
{
	d_ptr->newTask(std::forward<PackagedTask>(task));
}

} //namespace JXQ
